Flink: Transform INSERT as one DELETE following one INSERT if configure to use UPSERT #1996
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
rdblue
reviewed
Dec 28, 2020
| * All INSERT/UPDATE_AFTER events from input stream will be transformed to UPSERT events, which means it will | ||
| * DELETE the old records and then INSERT the new records. In partitioned table, the partition fields should be | ||
| * a subset of equality fields, otherwise the old row that located in partition-A could not be deleted by the | ||
| * new row that located in partition-B. |
Contributor
There was a problem hiding this comment.
Does anything validate this constraint?
rdblue
reviewed
Dec 28, 2020
| switch (row.getRowKind()) { | ||
| case INSERT: | ||
| case UPDATE_AFTER: | ||
| if (upsert) { |
Contributor
There was a problem hiding this comment.
It seems like this should only happen for the INSERT case because UPDATE_AFTER implies that there was an UPDATE_BEFORE that will perform the delete. This would delete the same row twice in that case, causing more equality deletes to be written for the row.
Contributor
|
Mostly looks good, but I don't think that upsert should be supported for |
coolderli
pushed a commit
to coolderli/iceberg
that referenced
this pull request
Apr 26, 2021
…re to use UPSERT (apache#1996)
Contributor
Contributor
Contributor
|
I think this is just waiting on someone to pick it up again. UPSERT should be unblocked now that row identifier fields have been added. |
Contributor
Member
Author
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
6 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Many people will export the result of flink aggregate values into apache iceberg table, for example:
This stream query will count the click number since the beginning of today (00:00:00), every emitted events will be a UPSERT events which overwrite the previous accumulated click_num.
In this cases, we will need to transform all INSERT/UPDATE_AFTER to be UPSERT, which means DELETE + INSERT the key.